package com.gvtv.main.execute;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.joda.time.DateTime;

import com.gvtv.main.db.ConnectionManager;
import com.gvtv.main.exception.ExecuteException;
import com.gvtv.main.execute.ExecuteJob.ExecuteJobState;
import com.gvtv.main.hive.HiveJdbc;
import com.gvtv.main.schedul.SchedulEnv;
import com.gvtv.main.schedul.SchedulInfo;
import com.gvtv.main.util.AppTools;
import com.gvtv.main.util.Constant;

public class Hive2MySqlJobWorker extends AbstractJobWorker implements JobWorker {

	private Logger logger = Logger.getLogger(Hive2MySqlJobWorker.class);

	private HiveJdbc hiveJdbc;
	private ExecutorService executorService;
	private volatile boolean isDone = false;
	private BlockingQueue<Object[]> queue;

	public Hive2MySqlJobWorker(SchedulInfo schedulInfo, ExecuteJob executeJob, SchedulEnv env) {
		super(schedulInfo, executeJob, env);
		this.hiveJdbc = new HiveJdbc();
		queue = new LinkedBlockingQueue<Object[]>(5000);
	}
	
	private void deleteData(){
		String delSql = AppTools.replaceBlank(job.getParamValue("delete"));
		if (StringUtils.isNotBlank(delSql)) {
			logger.info("job ["+job.getId()+"]删除数据语句为: " + delSql);
			QueryRunner queryRunner = new QueryRunner();
			Connection conn = null;
			try {
				conn = ConnectionManager.getConnection(job.getSchedulId() + "@" + job.getId(), job.getParamValue("driverClass"),
						job.getParamValue("jdbcUrl"), job.getParamValue("username"), job.getParamValue("password"));
				long delNums = queryRunner.update(conn, delSql);
				logger.info("job ["+job.getId()+"]删除了"+delNums+"数据.");
			} catch (SQLException e) {
				logger.error("job ["+job.getId()+"]删除数据失败."+e.getMessage());
			} catch (ExecuteException e) {
				logger.error("job ["+job.getId()+"]获取数据库连接失败."+e.getMessage());
			} finally {
				DbUtils.closeQuietly(conn);
			}
		}
	}

	@Override
	public void work() {
		job.setStartTime(new Date());
		logger.info("job ["+job.getId()+"]开始运行，运行时间："+new DateTime(job.getSubmitTime().getTime()).toString(Constant.DATETIME_FORMAT));
		env.convert(job);
		String query = AppTools.replaceBlank(job.getParamValue("query"));
		if (StringUtils.isBlank(query)) {
			logger.error("job ["+job.getId()+"]查询语句为空.");
			job.setJobState(ExecuteJobState.FAILED);
			return;
		}
		try {
			this.deleteData();
			int nums = StringUtils.isBlank(job.getParamValue("insertThreads")) ? 1 : Integer.valueOf(job.getParamValue("insertThreads"));
			if (nums <= 0) {
				logger.warn("job ["+job.getId()+"]insertThreads 参数值小于等于0,无效参数,设置默认值1 .");
				nums = 1;
			}
			List<String> sets = job.getHiveSets();
			if(sets!=null && !sets.isEmpty()){
				for (String set : sets) {
					if(StringUtils.isBlank(set)){
						logger.warn("job ["+job.getId()+"] 的hive set 语句为空，跳过。。。");
						continue;
					}
					hiveJdbc.execute(AppTools.replaceBlank(set));
				}
			}
			ResultSet rs = null;
			List<String> args = job.getArgs();
			if(args!=null && args.size()>0){
				rs = hiveJdbc.query(query, args.toArray(new String[]{}));
			}else {
				rs = hiveJdbc.query(query);
			}
			executorService = Executors.newFixedThreadPool(nums);
			List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>();
			for (int i = 0; i < nums; i++) {
				futures.add(executorService.submit(new MySqlWriter()));
			}
			ResultSetMetaData rsmd = rs.getMetaData();
			int rowCnt = rsmd.getColumnCount();
			long cnt = 0;
			while (rs.next()) {
				if (isDone()) {
					break;
				}
				Object[] data = new Object[rowCnt];
				for (int i = 0; i < rowCnt; i++) {
					data[i] = rs.getObject(i + 1);
				}
				queue.put(data);
				cnt++;
			}
			logger.info("job ["+job.getId()+"]查询了"+cnt+"条数据.");
			setDone(true);
			boolean success = true;
			for (Future<Boolean> future : futures) {
				if (!future.get().booleanValue()) {
					success = false;
					break;
				}
			}
			if (success) {
				job.setJobState(ExecuteJobState.SUCCESS);
			} else {
				job.setJobState(ExecuteJobState.FAILED);
				this.deleteData();
			}
		} catch (ExecuteException e) {
			setDone(true);
			job.setJobState(ExecuteJobState.FAILED);
			logger.error(e);
		} catch (SQLException e) {
			setDone(true);
			job.setJobState(ExecuteJobState.FAILED);
			logger.error(e);
		} catch (InterruptedException e) {
			setDone(true);
			job.setJobState(ExecuteJobState.FAILED);
			logger.error(e);
		} catch (ExecutionException e) {
			setDone(true);
			job.setJobState(ExecuteJobState.FAILED);
			logger.error(e);
		}
		job.setEndTime(new Date());
		logger.info("job ["+job.getId()+"]结束运行，结束时间："+new DateTime(job.getEndTime().getTime()).toString(Constant.DATETIME_FORMAT)+"，运行状态，"+job.getJobState());
	}

	@Override
	protected void clear() {
		hiveJdbc.close();
		if (executorService != null) {
			executorService.shutdown();
		}
	}

	public boolean isDone() {
		return isDone;
	}

	public void setDone(boolean isDone) {
		this.isDone = isDone;
	}

	public BlockingQueue<Object[]> getQueue() {
		return queue;
	}

	public void setQueue(BlockingQueue<Object[]> queue) {
		this.queue = queue;
	}

	public class MySqlWriter implements Callable<Boolean> {

		@Override
		public Boolean call() throws Exception {
			logger.info(job.getSchedulId()+"-"+job.getId()+" MySql数据库写入线程"+Thread.currentThread().getName()+"开始运行.");
			String insertSql = AppTools.replaceBlank(job.getParamValue("insert"));
			if (StringUtils.isBlank(insertSql)) {
				logger.error("job ["+job.getId()+"]插入语句为空.");
				setDone(true);
				return false;
			}
			logger.info("job ["+job.getId()+"]插入语句为: " + insertSql);
			int commitNum = 3000;
			if(StringUtils.isNumeric(job.getParamValue("commitNum"))){
				commitNum = Integer.valueOf(job.getParamValue("commitNum"));
			}
			boolean rel = true;
			long cnt = 0;
			Connection conn = null;
			PreparedStatement pst = null;
			try {
				conn = ConnectionManager.getConnection(job.getSchedulId() + "@" + job.getId(), job.getParamValue("driverClass"),
							job.getParamValue("jdbcUrl"), job.getParamValue("username"), job.getParamValue("password"));
				conn.setAutoCommit(false);
				pst = conn.prepareStatement(insertSql);
				while (true) {
					Object[] data = queue.poll(5, TimeUnit.SECONDS);
					if (data != null) {
						ConnectionManager.fillStatement(pst, data);
						pst.addBatch();
						cnt++;
					} else if (isDone() && queue.size()<=0) {
						break;
					}
					if(cnt%commitNum == 0){
						pst.executeBatch();
						conn.commit();
						logger.info(job.getSchedulId()+"-"+job.getId()+" 线程"+Thread.currentThread().getName()+"写入了"+cnt+"条数据.");
					}
				}
				pst.executeBatch();
				conn.commit();

			} catch (InterruptedException e) {
				conn.rollback();
				rel = false;
				setDone(true);
				logger.error("job ["+job.getId()+"]从队列中获取数据出错. "+e);
			} catch (SQLException e) {
				conn.rollback();
				rel = false;
				setDone(true);
				logger.error("job ["+job.getId()+"]写入数据到数据库出错. "+e);
			} finally {
				if(conn!=null){
					conn.setAutoCommit(true);
				}
				DbUtils.closeQuietly(conn, pst, null);
			}
			logger.info(job.getSchedulId()+"-"+job.getId()+" MySql数据库写入线程"+Thread.currentThread().getName()+"结束运行,写入了" + cnt + "条数据.");
			return rel;
		}

	}

}
